#ifndef __ASYNC_WORKER_H__
#define __ASTNC_WORKER_H__

#include <mutex>
#include <condition_variable>
#include <thread>
#include <functional>
#include <memory>
#include <atomic>
#include "../buffer/buffer.hpp"
#include "../util/handlerMessage.hpp"

// 异步日志工作器

namespace log_system
{
    using callbackFunction = std::function<void(log_system::Buffer &)>; // 回调函数

    class AsyncWorker
    {
    public:
        using pointer = std::shared_ptr<AsyncWorker>; // 本类指针
    public:
        AsyncWorker(const callbackFunction &callback_function)
            : _is_stop(false),
              _async_thread(std::thread(&AsyncWorker::routine, this)),
              _callback_function(callback_function) {}

        ~AsyncWorker()
        {
            stop();
        }

    public:
        void stop() // 停止日志工作器工作
        {
            _is_stop = true;
            _consumer_condition.notify_all(); // 唤醒所有消费缓冲区
            _async_thread.join();
        }

        void write(const char *data, size_t length) // 缓冲区中写入数据
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _producer_buffer.write(data, length);
            _consumer_condition.notify_one();
        }

    private:
        void routine() // 线程例程(消费缓冲区处理)
        {
            while (true)
            {
                { // 临界区
                    std::unique_lock<std::mutex> lock(_mutex);
                    // 生产缓冲区有数据则交换, 否则阻塞
                    _consumer_condition.wait(lock, [&]() -> bool
                                             { 
                                                // std::cout << "_consumer_condition wait!!!" << std::endl;
                                                return !_producer_buffer.empty() || _is_stop; });
                    // std::cout << "_consumer_condition not wait!!!" << std::endl;
                    if (_is_stop && _producer_buffer.empty())
                        break;
                    _consumer_buffer.swap(_producer_buffer);
                }
                _producer_condition.notify_one();     // 唤醒生产缓冲区
                _callback_function(_consumer_buffer); // 处理消费缓冲区
                _consumer_buffer.reset();
            }
        }

    private:
        std::atomic<bool> _is_stop;                  // 是否停止工作
        log_system::Buffer _producer_buffer;         // 生产缓冲区
        log_system::Buffer _consumer_buffer;         // 消费缓冲区
        std::mutex _mutex;                           // 互斥锁
        std::condition_variable _producer_condition; // 生产条件变量
        std::condition_variable _consumer_condition; // 消费条件变量
        std::thread _async_thread;                   // 异步工作线程
        callbackFunction _callback_function;         // 回调函数
    };

} // namespace log_system

#endif // #ifndef __ASYNC_WORKER_H__
